package cn.gwm.flink.streaming.dwd.standardization;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.functions.MapFunction;

import java.util.List;

/**
 * @ClassName DwdHiveField2DwdMapFunction
 * @Description dwd hive表字段 转换为可用字段
 * @Author LiangGuang
 * @Date 2023/04/26 15:43
 */
public class DwdHiveField2DwdMapFunction implements MapFunction<JSONArray, JSONObject> {

    private List<String> list;

    public DwdHiveField2DwdMapFunction(BeanSource.SourceEnum source) {
        this.list = BeanSource.sourceList(source);
    }

    @Override
    public JSONObject map(JSONArray objects) throws Exception {
        JSONObject result = JSONUtil.createObj();
        if (objects.size() > 0) {
            JSONObject temp = (JSONObject) objects.get(0);
            list.forEach(item -> {
                if (temp.containsKey(item.toLowerCase())) {
                    result.set(item, temp.get(item.toLowerCase()));
                    temp.remove(item.toLowerCase());
                } else {
                    result.set(item, BaseFields.defaultVal);
                }
            });
            temp.forEach((k, v) -> {
                result.set(k, v);
            });
        }
        return result;
    }

}
